package com.etl.service.impl;


import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.etl.constants.CommonConstant;
import com.etl.constants.UserSession;
import com.etl.entity.*;
import com.etl.enums.ResultEnum;
import com.etl.exception.GlobalException;
import com.etl.mapper.*;
import com.etl.model.rev.EtlTaskListRev;
import com.etl.model.rev.TableStructureRev;
import com.etl.service.IEtlTaskMainService;
import com.etl.service.KettleService;
import com.etl.util.*;
import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.logging.log4j.util.Strings;
import org.springframework.beans.BeanUtils;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import javax.websocket.Session;
import java.io.IOException;
import java.sql.*;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * Created by Administrator on 2019/10/21.
 */
@Service
@Slf4j
public class EtlTaskMainServiceImpl implements IEtlTaskMainService {


    @Resource
    private EtlTaskMainMapper etlTaskMainMapper;

    @Resource
    private EtlBusinessBatchMapper etlBusinessBatchMapper;

    @Resource
    private EtlBusinessDbInfoMapper dbInfoMapper;

    @Resource
    private EtlBusinessDbInfoMapper etlBusinessDbInfoMapper;

    @Resource
    private EtlTaskMainHistoryMapper etlTaskMainHistoryMapper;

    @Resource
    private EtlTaskBatchMapper etlTaskBatchMapper;

    @Resource
    private KettleService kettleService;



    @Override
    public PageInfo<EtlTaskMain> getAllTask(EtlTaskListRev etlTaskListRev) {
        PageHelper.startPage(etlTaskListRev.getPages(), etlTaskListRev.getSizes());
        QueryWrapper<EtlTaskMain> queryWrapper = new QueryWrapper<>();
        if (etlTaskListRev.getBatch() != null) {
            queryWrapper.eq("batch", etlTaskListRev.getBatch());
        }
        if (etlTaskListRev.getExecflag() != null) {
            queryWrapper.eq("exec_flag", etlTaskListRev.getExecflag());
        }
        if (Strings.isNotBlank(etlTaskListRev.getSrctable())) {
            queryWrapper.like("source_table_name", etlTaskListRev.getSrctable().trim());
        }

        //普通用户的话,只能查询自己的任务
        if(UserSession.getUser().getUserType() == 0){
            queryWrapper.eq("create_account",UserSession.getUser().getAccountName());
        }


        queryWrapper.orderByDesc("create_time");
        List<EtlTaskMain> etlTaskMains = etlTaskMainMapper.selectList(queryWrapper);
        PageInfo<EtlTaskMain> pageInfoDemo = new PageInfo<>(etlTaskMains);
        return pageInfoDemo;
    }

    @Override
    public void addTask(String sourceTableName, String querySql, Integer batch, String comment,Integer tableType) {

        /**
         * 查询批次号
         */
        EtlBusinessBatch etlBusinessBatch = etlBusinessBatchMapper.selectById(batch);
        if (etlBusinessBatch == null) {
            throw new GlobalException(ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST);
        }


        EtlBusinessDbInfo sourceEtlBusinessDbInfo = etlBusinessDbInfoMapper.selectById(etlBusinessBatch.getSourceDbId());
        if (sourceEtlBusinessDbInfo == null) {
            throw new GlobalException(ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST);
        }
        EtlBusinessDbInfo targetEtlBusinessDbInfo = etlBusinessDbInfoMapper.selectById(etlBusinessBatch.getTargetDbId());
        if (targetEtlBusinessDbInfo == null) {
            throw new GlobalException(ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST);
        }



        /**
         * 构建插入任务所需各个参数
         */
        // 批次号
        Integer batchNum = etlBusinessBatch.getBatch();


        //大写
        if(tableType == 2){
            sourceTableName = sourceTableName.toUpperCase();
        }else{
            sourceTableName = sourceTableName.toLowerCase();
        }

        String targetTableName = sourceTableName;
        String tableName1 = targetTableName + CommonConstant.NAMENEW.toLowerCase();      // 目标库中间表A
        String tableName2 = targetTableName + CommonConstant.NAMEOLD.toLowerCase();      // 目标库中间表B

        int commitSize = CommonConstant.COMMITSIZE;                 // 每多少行提交一次
        Integer execFlag = CommonConstant.EXECFLAG;                  // 任务执行状态 ，0任务运行成功等待下次执行，1任务运行中,2运行错误(问题修复后需手动修改为0程序才可再次执行)
        String sql1 = querySql;     // SQL1
        String sql2 = CommonConstant.ALERT + targetTableName + CommonConstant.RENAME + tableName2;
        String sql3 = CommonConstant.ALERT + tableName1 + CommonConstant.RENAME + targetTableName;
        String sql4 = CommonConstant.ALERT + tableName2 + CommonConstant.RENAME + tableName1;


        //目标数据库类型,设置表重命名语句
        String dbType = targetEtlBusinessDbInfo.getDbType();
        if(CommonConstant.MSSQL_NAME.equalsIgnoreCase(dbType)){
            sql2 = CommonConstant.EXEC_SP_RENAME + targetTableName + CommonConstant.COMMA + tableName2;
            sql3 = CommonConstant.EXEC_SP_RENAME + tableName1 + CommonConstant.COMMA + targetTableName;
            sql4 = CommonConstant.EXEC_SP_RENAME + tableName2 + CommonConstant.COMMA + tableName1;
        }



        // 构建实体对象
        EtlTaskMain etlTaskMain = new EtlTaskMain();
        etlTaskMain.setBatch(batchNum);
        etlTaskMain.setBatchName(etlBusinessBatch.getBatchName());

        etlTaskMain.setSourceDbId(etlBusinessBatch.getSourceDbId());
        etlTaskMain.setSourceDbSysname(sourceEtlBusinessDbInfo.getDbSysname());
        etlTaskMain.setSourceTableName(sourceTableName);


        etlTaskMain.setTargetDbId(etlBusinessBatch.getTargetDbId());
        etlTaskMain.setTargetDbSysname(etlBusinessBatch.getTargetDbSysname());
        etlTaskMain.setTargetTableName(targetTableName);


        etlTaskMain.setTableName1(tableName1);
        etlTaskMain.setTableName2(tableName2);
        etlTaskMain.setFlag(1);
        etlTaskMain.setCommitSize(commitSize);
        etlTaskMain.setExecFlag(execFlag);
        etlTaskMain.setSql1(sql1);
        etlTaskMain.setSql2(sql2);
        etlTaskMain.setSql3(sql3);
        etlTaskMain.setSql4(sql4);
        etlTaskMain.setTableComment(comment);

        etlTaskMain.setCreateAccount(UserSession.getUser().getAccountName());
        etlTaskMain.setCreateTime(new Date());
        etlTaskMain.setUpdateTime(new Date());
        //持久化
        etlTaskMainMapper.insert(etlTaskMain);

        log.info("新增任务:"+ etlTaskMain.getKid());
        //this.executeSingle(etlMainTask.getKid(),"创建表结构作业","创建表结构作业");

    }


    @Override
    public void updateTask(EtlTaskMain etlTaskMain) {
        etlTaskMainMapper.updateById(etlTaskMain);
        //创建表结构
//        this.executeSingle(etlMainTask.getKid(),"创建表结构作业","创建表结构作业");
    }




    public void executeSingle(Long kid,String kettleHome,String kettleJobName) {

        /**
         * 检验任务
         */
        QueryWrapper<EtlTaskMain> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("KID", kid);
        EtlTaskMain etlTaskMain = etlTaskMainMapper.selectOne(queryWrapper);
        if (etlTaskMain == null) {
            log.error("【执行单个任务】-{}", ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST.getErrMsg());
            throw new GlobalException(ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST);
        }
        if (etlTaskMain.getExecFlag() == 1) {
            log.error("【执行单个任务】-{}", ResultEnum.TASK_IS_RUNNING.getErrMsg());
            throw new GlobalException(ResultEnum.TASK_IS_RUNNING);
        }
        /**
         * 获取业务系统数据库连接
         */
        //来源数据库信息
        EtlBusinessDbInfo sourceDbInfo = dbInfoMapper.selectById(etlTaskMain.getSourceDbId());
        //目标数据库信息
        EtlBusinessDbInfo tgtDbInfo = dbInfoMapper.selectById(etlTaskMain.getTargetDbId());



        /**
         * 更新任务状态 execflag = 0 等待执行
         */
        EtlTaskMain update = new EtlTaskMain();
        QueryWrapper<EtlTaskMain> updateWraper = new QueryWrapper();
        updateWraper.eq("kid", kid);
        update.setExecFlag(0);
        etlTaskMainMapper.update(update, updateWraper);
        List<Long> kids = Lists.newArrayList();
        kids.add(kid);

        try {
            Map<String,String> jobVariableMap = new HashMap<>();
            jobVariableMap.put("kid",kid.toString());

//            jobVariableMap.put("tablename", etlTaskMain.getTargetTableName());
//            jobVariableMap.put("tablename1", etlTaskMain.getTableName1());


            kettleService.executeTask(jobVariableMap,kettleHome,kettleJobName,sourceDbInfo,tgtDbInfo, kids,"main");

        } catch (Exception e) {
            log.error("ETL错误通知", "【" + etlTaskMain.getSourceDbSysname() + "】ETL抽取任务失败", "【" + etlTaskMain.getSourceDbSysname() + "】抽取任务失败", "【" + etlTaskMain.getTargetTableName() + "】表任务失败");
            e.printStackTrace();
            throw new GlobalException(500, e.getMessage());
        }


    }



    @Async
    @Override
    public void socketExecuteSingle(Session session, Long kid) {
        try {
            executeSingle(kid,"单点全量抽取作业","单点全量抽取作业");
            session.getBasicRemote().sendText("etlTaskDone");
        } catch (Exception e) {
            try {
                session.getBasicRemote().sendText("etlTaskErr");
            } catch (IOException e1) {
                e1.printStackTrace();
            }
        }

    }



    /**
     * 执行批次下的所有任务
     *
     * @param batch
     */
    @Override
    public void executeAll(Integer batch, String accountName) {
        long beginTime = System.currentTimeMillis();
        /**
         * 查询批次下所有任务
         */
        QueryWrapper<EtlTaskMain> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("batch", batch);
        List<EtlTaskMain> etlTaskMains = etlTaskMainMapper.selectList(queryWrapper);
        if (etlTaskMains == null || etlTaskMains.size() == 0) {
            log.error("【执行批次下所有任务】-{}", ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST.getErrMsg());
            throw new GlobalException(ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST);
        }
        /**
         * 查询任务数据库信息
         */
//        QueryWrapper<EtlBusinessDbInfo> dbWraper = new QueryWrapper<>();
//        String sysName = etlMainTasks.get(0).getSourcedb();
//        dbWraper.eq("NAME", sysName);
//        EtlBusinessDbInfo dbInfo = dbInfoMapper.selectOne(dbWraper);



        //来源数据库信息
        EtlBusinessDbInfo sourceDbInfo = dbInfoMapper.selectById(etlTaskMains.get(0).getSourceDbId());
        String  sourceSysName = etlTaskMains.get(0).getSourceDbSysname();


        //目标数据库信息
        EtlBusinessDbInfo tgtDbInfo = dbInfoMapper.selectById(etlTaskMains.get(0).getTargetDbId());
        String  tgtSysName = etlTaskMains.get(0).getTargetDbSysname();


        String sysname = "来源数据库:"+sourceSysName+"抽取到目前库:"+tgtSysName;


        /**
         * 更新批次下任务状态
         */
        EtlTaskMain update = new EtlTaskMain();
        update.setExecFlag(0);
        queryWrapper.eq("flag", "1");
        queryWrapper.ne("exec_flag", "1");
        etlTaskMainMapper.update(update, queryWrapper);
        /**
         * 更新批次表状态1正在执行
         */
        EtlBusinessBatch etlBusinessBatch = new EtlBusinessBatch();
        etlBusinessBatch.setExecFlag(1);
        QueryWrapper<EtlBusinessBatch> wrapper = new QueryWrapper<>();
        wrapper.eq("batch", batch);
        etlBusinessBatchMapper.update(etlBusinessBatch, wrapper);
        //
        List<Long> kids = Lists.newArrayList();
        for (EtlTaskMain item : etlTaskMains) {
            kids.add(item.getKid());
        }
        try {


            Map<String,String>  jobVariableMap = new ConcurrentHashMap<>();
            jobVariableMap.put("batch", etlTaskMains.get(0).getBatch().toString());
            kettleService.executeTask(jobVariableMap,"全量抽取作业","全量抽取作业",sourceDbInfo,tgtDbInfo, kids ,"main");

        } catch (Exception e) {
            log.error("ETL错误通知", "【" + sysname + "】ETL抽取任务失败", "【" + sysname + "】抽取任务失败", "抽取该批次下所有任务失败");
            e.printStackTrace();
            throw new GlobalException(500, e.getMessage());
        }
        /**
         * 更新批次表状态 0:等待执行
         */
        etlBusinessBatch.setExecFlag(0);
        wrapper.eq("batch", batch);
        etlBusinessBatchMapper.update(etlBusinessBatch, wrapper);
        log.info("【{}】任务执行完成。耗时-{}秒", sysname, (System.currentTimeMillis() - beginTime) / 1000);
    }

    /**
     * 延时执行批次下的所有任务
     * 也就是将任务插入到定时任务表中，等待扫描器扫描并执行
     */
    @Override
    public void delayExecuteAll(Integer batch) {
        QueryWrapper<EtlBusinessBatch> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("batch", batch);
        EtlBusinessBatch etlBusinessBatch = etlBusinessBatchMapper.selectOne(queryWrapper);
        if (etlBusinessBatch == null) {
            throw new GlobalException(ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST);
        }
        EtlTaskBatch etlTaskBatch = new EtlTaskBatch();
        etlTaskBatch.setBatch(batch);
        etlTaskBatch.setTaskId(CodeGeneratedUtil.genUUID());
        etlTaskBatch.setWaitRunTime(new Date());
        etlTaskBatch.setAccountName(UserSession.getUser().getAccountName());
        etlTaskBatch.setCreateTime(new Date());
        etlTaskBatchMapper.insert(etlTaskBatch);
    }



    /**
     * 删除任务和表结构
     *
     * @param kid
     */
    @Override
    @Transactional
    public void removeTaskAndTable(Long kid, Integer type) {

        String accountName = UserSession.getUser().getAccountName();
        EtlTaskMain etlTaskMain = etlTaskMainMapper.selectById(kid);
        if (etlTaskMain == null) {
            log.error("【删除任务】-{}", ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST.getErrMsg());
            throw new GlobalException(ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST);
        }
        /**
         * 删除表
         */

        try {
            String tableName1 = etlTaskMain.getTargetTableName().trim();
            String tableName2 = etlTaskMain.getTableName1().trim();
            String delDate = DateUtil.getCurrent("yyMMddHHmmss");

            EtlBusinessDbInfo etlBusinessDbInfo = etlBusinessDbInfoMapper.selectById(etlTaskMain.getTargetDbId());
            //解密数据库密码
            etlBusinessDbInfo.setPassword(AESUtil.decrypt(etlBusinessDbInfo.getPassword(), AESUtil.password));

            if(CommonConstant.MSSQL_NAME.equalsIgnoreCase(etlBusinessDbInfo.getDbType())){
                //先判断是否存在该表，存在则删除
                String tableSql1 = "SELECT name tableName FROM sysobjects where name='"+tableName1+"'";
                String tabelName1 = DbUntil.getTabelName(tableSql1, etlBusinessDbInfo);
                if(StringUtils.isNotBlank(tabelName1)){
                    String deleteTable1 = "exec sp_rename "+tableName1+","+tableName1 + "_"+ accountName+"_DEL_" + delDate;
                    DbUntil.executeSql(deleteTable1,etlBusinessDbInfo);
                }

                //先判断是否存在该表，存在则删除
                String tableSql2 = "SELECT name tableName FROM sysobjects where name='"+tableName2+"'";
                String tabelName2 = DbUntil.getTabelName(tableSql2, etlBusinessDbInfo);
                if(StringUtils.isNotBlank(tabelName2)){

                    String deleteTable2 = "exec sp_rename "+tableName2+","+tableName2 + "_"+ accountName+"_DEL_" + delDate;
                    DbUntil.executeSql(deleteTable2,etlBusinessDbInfo);
                }

            }else{


                //先判断是否存在该表，存在则删除
                String tableSql1 = "";
                if(CommonConstant.ORACLE_NAME.equalsIgnoreCase(etlBusinessDbInfo.getDbType())){ //oracle
                    tableSql1 = "select table_name tableName from user_tables where table_name = '"+tableName1.toUpperCase()+"'";

                }else if(CommonConstant.MYSQL_NAME.equalsIgnoreCase(etlBusinessDbInfo.getDbType())){ //mysql
                    tableSql1 = "select table_name tableName from information_schema.TABLES  where table_name = '"+tableName1+"'" +"and table_schema='"+etlBusinessDbInfo.getDbSchema()+"'";
                }
                String tabelName1 = DbUntil.getTabelName(tableSql1, etlBusinessDbInfo);
                if(StringUtils.isNotBlank(tabelName1)){

                    String deleteTable1 = "ALTER TABLE "+tableName1+" RENAME TO "+tableName1 + "_"+ accountName+"_DEL_" + delDate;
                    DbUntil.executeSql(deleteTable1,etlBusinessDbInfo);
                }

                //先判断是否存在该表，存在则删除
                String tableSql2 = "";
                if(CommonConstant.ORACLE_NAME.equalsIgnoreCase(etlBusinessDbInfo.getDbType())){ //oracle
                    tableSql2 = "select table_name tableName from user_tables where table_name = '"+tableName2.toUpperCase()+"'";

                }else if(CommonConstant.MYSQL_NAME.equalsIgnoreCase(etlBusinessDbInfo.getDbType())){ //mysql
                    tableSql2 = "select table_name tableName from information_schema.TABLES where table_name = '"+tableName2+"'"+"and table_schema='"+etlBusinessDbInfo.getDbSchema()+"'";
                }
                String tabelName2 = DbUntil.getTabelName(tableSql2, etlBusinessDbInfo);
                if(StringUtils.isNotBlank(tabelName2)){

                    String deleteTable2 = "ALTER TABLE "+tableName2+" RENAME TO "+tableName2 + "_"+ accountName+"_DEL_" + delDate;
                    DbUntil.executeSql(deleteTable2,etlBusinessDbInfo);
                }

            }




//            String deleteTable3 = "ALTER TABLE "+tableName3+" RENAME TO "+tableName3 + "_"+ AccountName+"_DEL_" + delDate;
//            DbUntil.executeSql(deleteTable3,etlBusinessDbInfo);

        } catch (SQLException throwables) {
            throwables.printStackTrace();
        }

        /**
         * 删除任务
         */
        if (type != null && type == 1) {
            etlTaskMainMapper.deleteById(kid);
            /**
             * 将删除的任务插入到历史表里
             */
            EtlTaskMainHistory etlTaskMainHistory = new EtlTaskMainHistory();
            BeanUtils.copyProperties(etlTaskMain, etlTaskMainHistory);
            etlTaskMainHistory.setKid( kid );
            etlTaskMainHistoryMapper.insert(etlTaskMainHistory);
        }


    }








    /**
     * 查询返回结果集, 并且转换字段
     *
     * @param sql
     * @param dbInfo
     * @throws SQLException
     */
    @Override
    public Map<String, List<Map<String, Object>>> executeQuery(String sql, EtlBusinessDbInfo dbInfo) throws SQLException {
        PreparedStatement pstmt = null;
        ResultSet rs = null;
        ResultSetMetaData md = null;
        //解密数据库密码
        dbInfo.setPassword(AESUtil.decrypt(dbInfo.getPassword(), AESUtil.password));
        //数据结果集容器
        List<Map<String, Object>> list = Lists.newArrayList();
        // 表结构容器
        List<Map<String, Object>> structureList = Lists.newArrayList();
        Connection conn = null;
        try {

            // 替换换行符
//            sql = sql.replaceAll("(\\r\\n|\\n|\\n\\r|\\t)", " ");
            conn = DbUntil.getConn(dbInfo);
            pstmt = conn.prepareStatement(sql);

            //建立一个结果集，用来保存查询出来的结果
            rs = pstmt.executeQuery();

            // 获取键名
            md = rs.getMetaData();

//            // 获取行的数量
            int columnCount = md.getColumnCount();
            Boolean flag = true;
            while (rs.next()) {
                Map<String, Object> rowData = Maps.newHashMap();
                for (int i = 1; i <= columnCount; i++) {
//                    if (flag) {
//                        // 处理列类型、长度等信息
//                        structureList.add(oracleColumeInfo(md, i, dbInfo));
//                    }
                    //单独处理timestamp类型导致的json序列化问题
                    if ("timestamp".equalsIgnoreCase(md.getColumnTypeName(i))) {
                        if (rs.getTimestamp(i) != null) {
                            // 获取键名及值
                            rowData.put(md.getColumnName(i), rs.getTimestamp(i));
                        } else {
                            rowData.put(md.getColumnName(i), "");
                        }
                    } else {
                        if (rs.getObject(i) != null) {
                            // 获取键名及值
                            rowData.put(md.getColumnName(i), rs.getObject(i));
                        } else {
                            rowData.put(md.getColumnName(i), "");
                        }
                    }
                }
                list.add(rowData);
            }
        } catch (SQLException e) {
            log.error("【动态连接数据库】连接oracle出错");
            e.printStackTrace();
            throw new GlobalException(e.getErrorCode(), e.getMessage());
        } finally {
            md = null;
            DbUntil.close(conn,pstmt, rs);
        }
        Map<String, List<Map<String, Object>>> resultMap = null;
        resultMap = new HashMap<>();
        resultMap.put("data", list);
        resultMap.put("structure",null);
        return resultMap;
    }



    /**
     * 执行sql 不返回结果集
     *
     * @param sql
     * @param dbInfo
     * @throws SQLException
     */
    public void executeSql(String sql, EtlBusinessDbInfo dbInfo) throws SQLException {
        //解密数据库密码
        dbInfo.setPassword(AESUtil.decrypt(dbInfo.getPassword(), AESUtil.password));
        PreparedStatement pstmt = null;
        Connection conn;
        conn = DbUntil.getConn(dbInfo);
        try {
            if (conn != null) {
                pstmt = conn.prepareStatement(sql);
            }
            if (pstmt != null) {
                pstmt.execute();
            }
        } catch (SQLException e) {
            log.error(e.getMessage());
            throw new GlobalException(2000,e.getMessage());
        } finally {
            DbUntil.close(conn,pstmt, null);
        }
    }


    /**
     * 【接口】在源数据库中查询10条记录返回
     * @param tableName
     * @param batch
     * @return
     */
    @Override
    public Map<String, List<Map<String, Object>>> executeSql(String tableName, String batch) {

        final int resultSetCount = 20;

        /**
         * 处理 SQL 注入
         */
        boolean attack = RegularUtil.containsSqlInjection(tableName);
        if (attack) {
            log.error(ResultEnum.SQL_CHECK_ERROR.getErrMsg());
            throw new GlobalException(ResultEnum.SQL_CHECK_ERROR);
        }

        /**
         * 判断前端上传的是表名还是SQL 语句
         * 1、select 开头，并且紧接着空格，可判断字符串是可直接执行的sql。
         * 2、否则就是表名，则需要拼接sql 语句
         */
        tableName = tableName.trim();
        String tmpTblName = tableName.toLowerCase().replaceAll(";", "");
        String regex = "^select|with\\b.*?";
        // 创建 Pattern 对象
        Pattern r = Pattern.compile(regex);
        // 现在创建 matcher 对象
        Matcher m = r.matcher(tmpTblName);
        String exeSql;
        if (!m.find()) {
            exeSql = "SELECT * FROM " + tableName;
        } else {
            exeSql = tableName;
        }

        /**
         *  查询源数据库连接信息
         */
//        QueryWrapper<EtlBusinessDbInfo> queryWrapper = new QueryWrapper<>();
//        queryWrapper.eq("name", sysName);


        EtlBusinessBatch etlBusinessBatch = etlBusinessBatchMapper.selectById(batch);
        if (etlBusinessBatch == null) {
            log.error(ResultEnum.DB_SYSTEM_NAME_NOT_EXIST.getErrMsg() + "--" + batch);
            throw new GlobalException(ResultEnum.DB_SYSTEM_NAME_NOT_EXIST);
        }

        EtlBusinessDbInfo etlBusinessDbInfo = etlBusinessDbInfoMapper.selectById(etlBusinessBatch.getSourceDbId());
        if (etlBusinessDbInfo == null) {
            log.error(ResultEnum.DB_SYSTEM_NAME_NOT_EXIST.getErrMsg() + "--" + batch);
            throw new GlobalException(ResultEnum.DB_SYSTEM_NAME_NOT_EXIST);
        }
        if (etlBusinessDbInfo.getDbType().equals(CommonConstant.MYSQL_NAME)) {
            String mysqlQuery = "select a.* from (%s) a limit %s";
            exeSql = String.format(mysqlQuery, exeSql, resultSetCount);

        } else if (etlBusinessDbInfo.getDbType().equals(CommonConstant.MSSQL_NAME)) {
            String mssqlQuery = "select top %s a.* from (%s) a ";
            exeSql = String.format(mssqlQuery, resultSetCount, exeSql);
        } else {
            String oracleQuery = "select a.* from (%s) a WHERE ROWNUM <= %s";
            exeSql = String.format(oracleQuery, exeSql, resultSetCount);
        }

        /**
         * 连接对应的数据库，查询10条数据，返回给前端
         */
        try {
            Map<String, List<Map<String, Object>>> results = executeQuery(exeSql, etlBusinessDbInfo);
            return results;
        } catch (SQLException e) {
            log.error(ResultEnum.SQL_EXECUTE_ERROR.getErrMsg());
            throw new GlobalException(ResultEnum.SQL_EXECUTE_ERROR);
        }



    }




    @Override
    public void createTable(TableStructureRev tableStructure) {

        // 获取业务系统对应的批次号
        QueryWrapper<EtlBusinessBatch> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("name", tableStructure.getBatchName());
        EtlBusinessBatch etlBusinessBatch = etlBusinessBatchMapper.selectOne(queryWrapper);
        Integer batchNum = etlBusinessBatch.getBatch();
        if (batchNum == null) {
            log.error(ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST.getErrMsg());
            throw new GlobalException(ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST);
        }



//        kettleService.executeTask(sourceDbInfo,tgtDbInfo, null, kids, accountName,"单点全量抽取作业","单点全量抽取作业");

    }




    @Override
    public Integer findTableByTableName(Integer batch, String tableName) {
        /**
         * 查询批次号
         */
        LambdaQueryWrapper<EtlTaskMain> queryWrapper = Wrappers.lambdaQuery();
        queryWrapper.eq(EtlTaskMain::getBatch, batch);
        queryWrapper.eq(EtlTaskMain::getTargetTableName,tableName);
        EtlTaskMain etlTaskMain = etlTaskMainMapper.selectOne(queryWrapper);
        if (etlTaskMain != null) {
            throw new GlobalException(ResultEnum.BATCH_EXIST_TABLE);
        }
        Integer count = 0;// etlBusinessDbInfoMapper.findTableByTableName(tgtTable.trim());
        return count;
    }



}

